package com.itcast.flink.daily;

import lombok.SneakyThrows;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 案例演示：每日实时大屏统计，模拟电商交易订单数据实时总销售额统计，实时大屏每隔1秒刷新一次
 * 方案：滚动窗口+触发器
 *
 * @author lilulu
 */
public class StreamDailyOrderReport {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        /**
         * 2023-04-06 23:59:01,order_10001,user_1,10.00
         * 2023-04-06 23:59:04,order_10002,user_2,10.00
         * 2023-04-06 23:59:58,order_10003,user_4,20.00
         * 2023-04-07 00:00:01,order_00001,user_2,100.00
         * 2023-04-07 00:00:12,order_00002,user_4,99.00
         */
        DataStreamSource<String> inputStream = env.socketTextStream("node1", 9999);
        // 3. 数据转换-transformation
        /**
         * 指定数据中事件时间字段的值，必须为Long类型
         * 提取字段值，封装到二元组
         * 设置滚动事件时间窗口进行计算
         */
        SingleOutputStreamOperator<Tuple2<String, Double>> mapStream = inputStream.filter(line -> line.trim().split(",").length == 4)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                                    private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

                                    @SneakyThrows
                                    @Override
                                    public long extractTimestamp(String element, long recordTimestamp) {
                                        System.out.println("order -> " + element);
                                        String orderTime = element.split(",")[0];
                                        return format.parse(orderTime).getTime();
                                    }
                                })
                ).map(
                        new MapFunction<String, Tuple2<String, Double>>() {
                            @Override
                            public Tuple2<String, Double> map(String value) throws Exception {
                                String money = value.split(",")[3];
                                return Tuple2.of("全国订单金额", Double.parseDouble(money));
                            }
                        }
                );
        SingleOutputStreamOperator<String> windowStream = mapStream.keyBy(tuple -> tuple.f0)
                .window(TumblingEventTimeWindows.of(Time.days(1),Time.hours(-8)))
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)))
                .apply(new WindowFunction<Tuple2<String, Double>, String, String, TimeWindow>() {
                    private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

                    @Override
                    public void apply(String key, TimeWindow timeWindow, Iterable<Tuple2<String, Double>> iterable, Collector<String> collector) throws Exception {
                        String startTime = this.format.format(timeWindow.getStart());
                        String endTime = this.format.format(timeWindow.getEnd());
                        double sum = 0.0;
                        for (Tuple2<String, Double> tuple2 : iterable) {
                            sum += tuple2.f1;
                        }
                        String output = "window[" + startTime + " ~ " + endTime + "], " + key + " = " + sum;
                        collector.collect(output);
                    }
                });
        // 4. 数据终端-sink
        windowStream.printToErr();
        // 5. 触发执行-execute
        env.execute("StreamDailyOrderReport");
    }
}